package com.lzq.mq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.ServiceRegistryRespDTO;
import com.lzq.mq.common.dto.SlaveAckDTO;
import com.lzq.mq.common.enums.NameServerResponseCode;
import com.lzq.mq.common.event.Listener;
import com.lzq.mq.nameserver.common.CommonCache;
import com.lzq.mq.nameserver.event.model.SlaveReplicationMsgAckEvent;

/**
 * 主从复制模式下, 主节点处理从节点返回ack消息的处理器
 * 用于主节点监听下游节点的同步情况, 根据同步情况返回ack给客户端
 */
public class ReplicationMasterReciveSlaveAckListener implements Listener<SlaveReplicationMsgAckEvent> {

    @Override
    public void onReceive(SlaveReplicationMsgAckEvent event) throws Exception {
        String slaveAckMsgId = event.getMsgId();
        SlaveAckDTO slaveAckDTO = CommonCache.getAckMap().get(slaveAckMsgId);
        if (slaveAckDTO == null) {
            return;
        }
        int currentAckTime = slaveAckDTO.getNeedAckTime().decrementAndGet();        // 下游有几个从节点, 就需要ack就
        //如果是复制模式，代表所有从节点已经ack完毕了，
        //如果是半同步复制模式
        if (currentAckTime == 0) {
            // 避免并发情况下，重复删除
            CommonCache.getAckMap().remove(slaveAckMsgId);
            // 生成消息体
            ServiceRegistryRespDTO respDTO = new ServiceRegistryRespDTO();
            respDTO.setMsgId(event.getMsgId());     // todo 这个msgId应该是client发送的!!
            byte[] respJsonBytes = JSON.toJSONBytes(respDTO);

            TcpMsg tcpMsg = new TcpMsg(
                    NameServerResponseCode.REGISTRY_SERVICE_SUCCESS.getCode(),
                    respJsonBytes
            );
            // 发送消息给Broker
            slaveAckDTO.getBrokerChannel().writeAndFlush(tcpMsg);
        }
    }
}
